package dnet

import (
	"dgo/goutils/utils"
	"fmt"
	"net"
	"sync"
	"sync/atomic"
	"time"
)

type ProxyTcpV2Item struct {
	frontend_sendsize int64
	frontend_recvsize int64
	backend_recvsize  int64
	disFlag           int32
	owner             *ProxyTcpV2
	StartT            time.Time
	FrontendAddr      string
	BackendAddr       string

	frontend_conn net.Conn
	backend_conn  net.Conn

	Tag       interface{}
	DisReason string
	sessionid string
}

func (this *ProxyTcpV2Item) GetSessionId() string {
	return this.sessionid
}

func (this *ProxyTcpV2Item) GetBackendInfo() string {
	return fmt.Sprintf("%s,%s", this.BackendAddr, utils.HumanFilesize(this.backend_recvsize))
}

func (this *ProxyTcpV2Item) GetFrontendInfo() string {
	return fmt.Sprintf("%s,%s", this.FrontendAddr, utils.HumanFilesize(this.frontend_sendsize))
}

func (this *ProxyTcpV2Item) GetBackendRecvSize() int64 {
	return this.backend_recvsize
}

func (this *ProxyTcpV2Item) GetFrontendRecvSize() int64 {
	return this.frontend_recvsize
}

func (this *ProxyTcpV2Item) GetFrontendSendSize() int64 {
	return this.frontend_sendsize
}

func (this *ProxyTcpV2Item) BackendWrite(buf []byte) (n int, err error) {
	conn := this.backend_conn
	if conn != nil {
		n, err = conn.Write(buf)
		if err == nil {
			owner := this.owner
			if owner != nil {
				evt := owner.OnBackendAfterWriteBuf
				if evt != nil {
					evt(this, buf)
				}
			}
		}
		return
	}
	return
}

func (this *ProxyTcpV2Item) FrontedWrite(buf []byte) (n int, err error) {
	conn := this.frontend_conn
	if conn != nil {
		n, err = conn.Write(buf)
		this.frontend_sendsize += int64(n)
		if err == nil {
			owner := this.owner
			if owner != nil {
				evt := owner.OnFrontendAfterWriteBuf
				if evt != nil {
					evt(this, buf)
				}
			}
		}
		return
	}
	return
}

func (this *ProxyTcpV2Item) RequestDisconnect(reason string) {
	if atomic.CompareAndSwapInt32(&this.disFlag, 0, 1) {
		this.DisReason = reason

		if this.frontend_conn != nil {
			this.frontend_conn.Close()
		}

		if this.backend_conn != nil {
			this.backend_conn.Close()
		}
	}

}

/*
*

	代理
*/
type ProxyTcpV2 struct {
	OnlineCnt        int32
	terminated       bool
	closeW           *sync.WaitGroup
	listener         *NetListener
	frontend_addr    string
	frontend_network string
	backend_addr     string
	Sessions         utils.GSyncMapGroup
	OnListenSucc     func()

	/**
	  获取后端连接, 可以返回空连接，表示暂时不进行连接
	*/
	OnDialBackend func(sender *ProxyTcpV2Item) (conn net.Conn, err error)

	/*
	  在doIoCopy的主线程
	*/
	OnBackendConnected func(sender *ProxyTcpV2Item)

	OnFrontendBuf           func(sender *ProxyTcpV2Item, buf []byte) (newBuf []byte)
	OnFrontendAfterWriteBuf func(sender *ProxyTcpV2Item, buf []byte)
	OnBackendBuf            func(sender *ProxyTcpV2Item, buf []byte) (newBuf []byte)
	OnBackendAfterWriteBuf  func(sender *ProxyTcpV2Item, buf []byte)
	OnConnected             func(sender *ProxyTcpV2Item)
	OnDisconnected          func(sender *ProxyTcpV2Item)
}

func NewProxyTcpV2() *ProxyTcpV2 {
	rval := &ProxyTcpV2{
		closeW:     new(sync.WaitGroup),
		terminated: true,
		listener:   NewNetListener(),
	}
	return rval
}

func (this *ProxyTcpV2) CheckDialBackend(itm *ProxyTcpV2Item) bool {
	if itm.backend_conn == nil {
		evt := this.OnDialBackend
		if evt != nil {
			conn, err := evt(itm)
			if err != nil {
				itm.RequestDisconnect(fmt.Sprintf("拨号后端出现异常:%s", err.Error()))
				return false
			}
			if conn != nil {
				itm.BackendAddr = conn.RemoteAddr().String()
				itm.backend_conn = conn
				evt0 := this.OnBackendConnected
				if evt0 != nil {
					evt0(itm)
				}
			}
		} else {
			conn, err := net.Dial("tcp4", this.backend_addr)
			if err != nil {
				itm.RequestDisconnect(fmt.Sprintf("拨号后端(%s)出现异常:%s", this.backend_addr, err.Error()))
				return false
			}
			if conn != nil {
				itm.BackendAddr = conn.RemoteAddr().String()
				itm.backend_conn = conn
				evt0 := this.OnBackendConnected
				if evt0 != nil {
					evt0(itm)
				}
			}
		}
	}
	return itm.backend_conn != nil
}

func (this *ProxyTcpV2) DoIoCopy(itm *ProxyTcpV2Item) {
	frontend := itm.frontend_conn

	this.CheckDialBackend(itm)
	if atomic.LoadInt32(&itm.disFlag) == 1 { // 直接断线了
		return
	}

	/**
	  处理后端数据转发
	*/
	go func() {
		if utils.GoFunCatchException {
			defer utils.PanicHandler()
		}
		buf := make([]byte, 1024)
		backend := itm.backend_conn
		for atomic.LoadInt32(&itm.disFlag) == 0 {
			if backend == nil {
				// 此处不进行创建, 统一由前端接收线程去创建
				backend = itm.backend_conn
				if backend == nil {
					time.Sleep(time.Millisecond * 100)
					continue
				}
			}
			n, err := backend.Read(buf)
			if err != nil {
				itm.RequestDisconnect(fmt.Sprintf("读取后端数据异常:%s", err.Error()))
				break
			}
			itm.backend_recvsize += int64(n)
			newBuf := buf[:n]
			evt := this.OnBackendBuf
			if evt != nil {
				newBuf = evt(itm, newBuf)
			}
			if len(newBuf) > 0 {
				_, err := itm.FrontedWrite(newBuf)
				if err != nil {
					itm.RequestDisconnect(fmt.Sprintf("发送数据到前端时出现异常:%s", err.Error()))
					break
				}

			}
		}
	}()

	/**
	  处理前端数据接收
	*/
	buf := make([]byte, 1024)
	for {
		n, err := frontend.Read(buf)
		if err != nil {
			itm.RequestDisconnect(fmt.Sprintf("读取前端数据异常:%s", err.Error()))
			break
		}
		itm.frontend_recvsize += int64(n)
		newBuf := buf[:n]
		evt := this.OnFrontendBuf
		if evt != nil {
			newBuf = evt(itm, newBuf)
		}
		if len(newBuf) > 0 {
			if this.CheckDialBackend(itm) {
				_, err := itm.BackendWrite(newBuf)
				if err != nil {
					itm.RequestDisconnect(fmt.Sprintf("发送数据到后端时出现异常:%s", err.Error()))
					break
				}
			}
		}
	}
}

func (this *ProxyTcpV2) handleFrontendConn(frontend_conn net.Conn) {
	if utils.GoFunCatchException {
		defer utils.PanicHandler()
	}
	atomic.AddInt32(&this.OnlineCnt, 1)
	defer atomic.AddInt32(&this.OnlineCnt, -1)
	itm := &ProxyTcpV2Item{
		owner:         this,
		sessionid:     fmt.Sprintf("%s_%d", utils.RandKeyString(12), utils.NewObjectDNA()),
		StartT:        time.Now(),
		frontend_conn: frontend_conn,
		FrontendAddr:  frontend_conn.RemoteAddr().String(),
		backend_conn:  nil,
	}
	this.Sessions.Set(itm.sessionid, itm)
	if this.OnConnected != nil {
		this.OnConnected(itm)
	}
	this.DoIoCopy(itm)
	this.onClose(itm)
	itm.owner = nil
}

func (this *ProxyTcpV2) onClose(rec *ProxyTcpV2Item) {
	this.Sessions.Remove(rec.sessionid)
	if this.OnDisconnected != nil {
		this.OnDisconnected(rec)
	}
}

func (this *ProxyTcpV2) GetBackendAddr() string {
	return this.backend_addr
}

func (this *ProxyTcpV2) GetFrontendAddr() string {
	return this.frontend_addr
}

func (this *ProxyTcpV2) Close() {
	this.terminated = true
	this.listener.Close()
}

func (this *ProxyTcpV2) GetActive() bool {
	return !this.terminated
}

func (this *ProxyTcpV2) BreakConnect() {
	this.listener.BreakConnect()
}

func (this *ProxyTcpV2) Open() {
	this.terminated = false
	this.closeW.Add(1)
	defer this.closeW.Done()
	this.listener.Config(this.frontend_network, this.frontend_addr)
	this.listener.OnListenSucc = func(network, addr string) {
		evt := this.OnListenSucc
		if evt != nil {
			evt()
		}
	}
	this.listener.HandleConnFunc = func(conn net.Conn) {
		go this.handleFrontendConn(conn)
	}
	this.listener.Open()
}

func (this *ProxyTcpV2) SetFrontendAddr(network, addr string) (ischanged bool) {
	ischanged = false
	if this.frontend_network != network {
		this.frontend_network = network
		ischanged = true
	}

	if this.frontend_addr != addr {
		this.frontend_addr = addr
		ischanged = true
	}
	return ischanged
}

func (this *ProxyTcpV2) SetBackendAddr(val string) bool {
	if this.backend_addr != val {
		this.backend_addr = val
		return true
	}
	return false
}
